package com.mng.processFun;

import com.mng.dto.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class KeyedProcessFunTest {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> stringDS = env.socketTextStream("localhost", 7777);
        SingleOutputStreamOperator<SensorReading> sensorDs = stringDS.map(new MapFunction<String, SensorReading>() {
            @Override
            public SensorReading map(String s) throws Exception {
                String[] strings = s.split(",");
                return new SensorReading(strings[0], Long.parseLong(strings[1]), Double.parseDouble(strings[2]));
            }
        });

        sensorDs.keyBy(sensorReading -> sensorReading.getId())
                .process(new MyProcess())
                .print();
        env.execute();
    }
}

class MyProcess extends KeyedProcessFunction<String,SensorReading,String>{
    @Override
    public void processElement(SensorReading sensorReading, Context context, Collector<String> collector) throws Exception {
        collector.collect(context.getCurrentKey());
        System.out.println(context.timerService().currentProcessingTime());
        context.timerService().registerProcessingTimeTimer(context.timerService().currentProcessingTime()+5000L);
    }
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
        super.onTimer(timestamp, ctx, out);
        out.collect(String.valueOf(timestamp));
    }
}